package cn.smileyan.demos.io;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;

import java.nio.charset.StandardCharsets;


/**
 * 将字节码数据进行序列化，以及将实体类转换
 * @author smileyan
 * @param <O> 实体类
 */
@Slf4j
public class CommonEntitySchema<O> implements DeserializationSchema<O>, SerializationSchema<O> {

    private final Class<O> clazz;

    public CommonEntitySchema(Class<O> clazz) {
        this.clazz = clazz;
    }

    @Override
    public O deserialize(byte[] message) {
        try {
            String str = new String(message, StandardCharsets.UTF_8);
            log.info("kafka received message: {}", str);
            return JSON.parseObject(str, clazz);
        } catch (Exception e) {
            log.error(e.getMessage());
        }
        return null;
    }

    @Override
    public boolean isEndOfStream(O nextElement) {
        return false;
    }

    @Override
    public TypeInformation<O> getProducedType() {
        return TypeInformation.of(clazz);
    }

    @Override
    public byte[] serialize(O element) {
        return JSON.toJSONBytes(element);
    }
}
